package com.zhang.hadoop.flink.test3;

import com.zhang.hadoop.flink.base.Event;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @author: zhang yufei
 * @createTime:2022/5/15 10:12
 * @description:
 */
public class TransformFlatMapTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<Event> events = new ArrayList<>();
        events.add(new Event("yanghui", "./yindao", 1000L));
        events.add(new Event("yuping", "./siwajiao", 2000L));
        events.add(new Event("yangdan", "./gangmen", 2000L));
        events.add(new Event("jingru", "./niaodao", 2000L));
        DataStreamSource<Event> stream = env.fromCollection(events);

        //进行转换计算，提取user字段

        //1.使用自定义类，实现MapFunction接口
        SingleOutputStreamOperator<String> result1 = stream.flatMap(new MyFlatMap());
        result1.print("result1");

        //2.使用匿名类实现MapFunction接口
        SingleOutputStreamOperator<String> result2 = stream.flatMap(new FlatMapFunction<Event, String>() {
            @Override
            public void flatMap(Event event, Collector<String> collector) throws Exception {
                collector.collect(event.user);
                collector.collect(event.url);
                collector.collect(event.timestamp.toString());
            }
        });
        result2.print("result2");

        //3.传入lambda表达式
        SingleOutputStreamOperator<String> result3 = stream.flatMap((Event event, Collector<String> collector) -> {
            if (event.user.equals("yuping")) {
                collector.collect(event.url);
            } else if (event.user.equals("yuping")) {
                collector.collect(event.user);
                collector.collect(event.url);
                collector.collect(event.timestamp.toString());
            }
        }).returns(String.class);
        result3.print("result3");

        env.execute();
    }

    //自定义MapFunction
    public static class MyFlatMap implements FlatMapFunction<Event, String> {

        @Override
        public void flatMap(Event event, Collector<String> collector) throws Exception {
            collector.collect(event.user);
            collector.collect(event.url);
            collector.collect(event.timestamp.toString());
        }
    }
}
